#include "Slave.h"

Slave::Slave(){
	
}

void Slave::SetParameters(int _slaveId, int _receiveBufferSize)
{
	this->slaveId = _slaveId;
	this->receiveBufferSize = _receiveBufferSize;
	this->receiveBuffer = new char[receiveBufferSize];
}

void Slave::Start()
{
	if (this->receiveBufferSize==0){
		printf("\033[44m->Rank%d: Call SetParameters() method first\033[0m\n", slaveId);
		return;
	}
	
	MPI_Request request;
	MPI_Status status;
	lastMessageFrom = -1;
	
	for(;;){
		printf("\033[44m->Rank%d: Receiving \033[0m\n", this->slaveId);
		int flag;
		int probeFrom;
		
		for(;;){
			MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, &status);
			if (flag==1){
				probeFrom = status.MPI_SOURCE;
				break;
			}
		}
		
		MPI_Irecv (receiveBuffer, receiveBufferSize, MPI_CHAR, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD,
					   &request);
		MPI_Wait (&request, &status);
		
		lastMessageFrom = status.MPI_SOURCE;
		
		// printf("\033[44m->Rank%d: Received from %d\033[0m\n", this->slaveId, lastMessageFrom);
		
		switch(status.MPI_TAG)
		{
				default:
					printf("\033[44m->Rank%d: Received unknown TAG received from %d\033[0m\n", slaveId, lastMessageFrom);
					break;
				
				case TAG_TERMINATE_SLAVE:
					printf("\033[44m->Rank%d: Received TAG_TERMINATE_SLAVE from %d\033[0m\n", slaveId, lastMessageFrom);
					return;
					
				case TAG_JOB:
					printf("\033[44m->Rank%d: Received TAG_JOB from %d\033[0m\n", slaveId, lastMessageFrom);
					CreateThread();
					break;
					
				case TAG_SYNC:
					printf("\033[44m->Rank%d: Received TAG_SYNC from %d\033[0m\n", slaveId, lastMessageFrom);
					SyncThreads();
					break;
					
				case TAG_CLOSE_ALL_THREADS:
					printf("\033[44m->Rank%d: Received TAG_CLOSE_ALL_THREADS from %d\033[0m\n", slaveId, lastMessageFrom);
					CloseAllThreads();
					break;
					
				case TAG_RESET_BUFFER_SIZE:
					printf("\033[44m->Rank%d: Received TAG_RESET_BUFFER_SIZE from %d\033[0m\n", slaveId, lastMessageFrom);
					ResetBufferSize();
					break;
					
				case TAG_SET_BUFFER_SIZE:
					printf("\033[44m->Rank%d: Received TAG_SET_BUFFER_SIZE from %d\033[0m\n", slaveId, lastMessageFrom);
					SetBufferSize();
					break;
		}
	}
}

void Slave::SyncThreads()
{
	int count = threads.size();
	if (count>0){
		for(int i=0; i<count; i++){
			pthread_join(threads[i], NULL);
			threads.erase(threads.begin()+i-1);
		}
		printf("\033[44m->Rank%d: SyncThreads() complete\033[0m\n", slaveId);
	}
	else
		printf("\033[44m->Rank%d: SyncThreads() no threads to sync\033[0m\n", slaveId);
	
}

void Slave::CloseAllThreads()
{
	int count = threads.size();
	if (count>0){
		for(int i=0; i<count; i++){
			pthread_cancel(threads[i]);
			threads.erase(threads.begin()+i-1);
		}
		printf("\033[44m->Rank%d: Closed all threads\033[0m\n", slaveId);
	}
	else{
		printf("\033[44m->Rank%d: No threads to close\033[0m\n", slaveId);
	}
	
}

void Slave::CreateThread()
{
	printf("\033[44m->Rank%d: Creating thread...\033[0m\n", slaveId);
	
	// stworzenie kopii obiektu slave, aby nowo utworzony watek posiadal 
	// dane, ktore nie zostana przez przypadek zmodyfikowane przez watek glowny
	Slave* slaveCopy = this->SlaveCopy();
	
	pthread_t thread;	
	
	pthread_create( &thread, NULL, Slave::callWorkerThread, (void*)slaveCopy);
	
	threads.push_back(thread);
}

Slave* Slave::SlaveCopy()
{
	Slave* slaveCopy = new Slave();
	memcpy(slaveCopy, this, sizeof(Slave));
	
	char* bufferCopy = new char[receiveBufferSize];
	memcpy(bufferCopy, receiveBuffer, receiveBufferSize);
	
	slaveCopy->receiveBuffer = bufferCopy;
	
	return slaveCopy;
}

// Metoda uruchomiona przez hack (funkcje callWorkerThread)
void* Slave::WorkerThread(Slave* slave)
{
	memcpy(this, slave, sizeof(Slave));
	printf("\033[44m->Rank%d: Message from thread created by job from node %d\033[0m\n", slaveId, lastMessageFrom);
	
	// wykonanie operacji uzytkownika klasy
	JobHandler(this->receiveBuffer, this->receiveBufferSize);

	// odeslanie wynikow do lastMessageFrom
	printf("\033[44m->Rank%d: Sending results to node %d\033[0m\n", slaveId, lastMessageFrom);
	
	MPI_Request request;
	MPI_Status status;
	
	MPI_Isend (receiveBuffer, receiveBufferSize, MPI_CHAR,
					   lastMessageFrom, TAG_JOB, MPI_COMM_WORLD, &request);					   
	MPI_Wait (&request, &status);
	
	printf("\033[44m->Rank%d: Results sent\033[0m\n", slaveId);
	
	// delete slave;
}

void Slave::ResetBufferSize()
{
	printf("\033[44m->Rank%d: ResetBufferSize()\033[0m\n", slaveId);
	this->receiveBufferSize = 1024;
	delete this->receiveBuffer;
	
	this->receiveBuffer = new char[1024];
}

void Slave::SetBufferSize()
{	
	int newBufferSize = 0;
	memcpy(&newBufferSize, receiveBuffer, 4);
	
	if (newBufferSize<=0) {
		printf("\033[44m->Rank%d: Cannot set bufferSize to %d bytes\n", slaveId, newBufferSize);
		return;
	}
	
	printf("\033[44m->Rank%d: SetBufferSize() to %d bytes\033[0m\n", slaveId, newBufferSize);
	this->receiveBufferSize = newBufferSize;
	delete this->receiveBuffer;
	this->receiveBuffer = new char[newBufferSize];
}

Slave::~Slave(){
	delete this->receiveBuffer;
}
